package org.springblade.mqtt;

import org.eclipse.paho.client.mqttv3.*;
import org.springblade.common.config.MqQueueConfig;
import org.springframework.stereotype.Component;

/**
 * MQTT处理器
 * @author 李家民
 */
@Component
public class MqttHandle implements MqttCallbackExtended {

	/**
	 * 连接丢失
	 * @param cause
	 */
	@Override
	public void connectionLost(Throwable cause) {
		System.out.println("connection lost：" + cause.getMessage());
		MqttInitialized.setConnectStatus(MqttInitialized.getClient().isConnected());
	}

	/**
	 * 收到消息
	 * @param topic
	 * @param message
	 */
	@Override
	public void messageArrived(String topic, MqttMessage message) {
		System.out.println("Received message: \n  topic：" + topic
			+ "\n  Qos：" + message.getQos()
			+ "\n  payload：" + new String(message.getPayload()));
	}

	/**
	 * 消息传递成功
	 * @param token
	 */
	@Override
	public void deliveryComplete(IMqttDeliveryToken token) {
		System.out.println("消息传递成功---deliveryComplete:" + token.getMessageId());
	}

	/**
	 * 连接成功
	 * @param reconnect If true, the connection was the result of automatic reconnect.
	 * @param serverURI The server URI that the connection was made to.
	 */
	@Override
	public void connectComplete(boolean reconnect, String serverURI) {
		System.out.println(reconnect + ":MQTT连接成功:" + serverURI);
		subscribe(MqQueueConfig.MQTT_TOPIC_DEMO, 0);
	}


	/**
	 * 消息订阅
	 * @param topic 主题
	 * @param qos   QOS
	 */
	public static void subscribe(String topic, Integer qos) {
		try {
			MqttInitialized.getClient().subscribe(topic, qos);
		} catch (MqttException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * 消息发布
	 * @param topic   主题
	 * @param message 消息体
	 * @param qos     QOS
	 */
	public static void publish(String topic, String message, Integer qos) {
		MqttMessage mqttMessage = new MqttMessage(message.getBytes());
		mqttMessage.setQos(qos);
		try {
			MqttInitialized.getClient().publish(topic, mqttMessage);
		} catch (MqttException e) {
			throw new RuntimeException(e);
		}
	}

}
